/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.pig.pigpen.editor.model;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.pig.pigpen.LoggerUtil;
import org.apache.pig.pigpen.editor.parser.PigPenParser;
import org.eclipse.core.runtime.IProgressMonitor;

public class Query extends PropertyAwareObject {

    public static final String LINK = "LINK";
    public static final String HEADER = "HEADER";

    int MAX_HEADER_ITEMS = 100;

    private File file = null;

    // the components of the query:
    private List<QueryComponent> components;

    // the parser for this query
    // (parsing is done in an incremental fashion, as the user edits the query)
    PigPenParser parser;

    // intialize an empty query
    public Query() {
        components = new ArrayList<QueryComponent>();
    }

    // read a query out of a .pig file
    public void loadFile(File file) throws IOException {
        this.file = file;

        BufferedReader reader = null;
        try {
            reader = new BufferedReader(new FileReader(file));
        } catch (IOException e) {
            throw new IOException("Error reading file" + file + ": "
                    + e.getMessage());
        }

        parser = new PigPenParser(reader);
        if (!parser.PigServerStart) {
            parser = null;
            return;
        }
        parser.setQuery(this);
        parser.setInteractive(false);

        boolean dupAliasPresent = parser.parseQuery();

        try {
            reader.close();
        } catch (IOException e) {
            LoggerUtil.logError("Error closing file " + file, e);
        }

        if (dupAliasPresent) {
            throw new IOException(
                    "Your pig script contains duplicate aliases. Pig Pen does not permit duplicate aliases. Please rename them so that there are no duplicates, before opening with the Pig Latin editor.");
        }

        padHeader();
    }

    public void reParseOp(Operator op) {

        if (parser == null) {
            parser = new PigPenParser();
            parser.setQuery(this);
            parser.setInteractive(false);
        }

        parser.reParseOp(op);
    }

    public void addComponent(QueryComponent component) {
        if (component instanceof Operator) {
            addOperator((Operator) component);
        } else {
            components.add(component);
            firePropertyChange(CHILD, null, component);
        }
    }

    public void removeComponent(QueryComponent component) {
        if (component instanceof Operator) {
            removeOperator((Operator) component);
        } else {
            components.remove(component);
            firePropertyChange(CHILD, component, null);
        }
    }

    public void removeAllComponents() {
        while (!components.isEmpty())
            removeComponent(components.get(0));
    }

    public void addSetting(String key, String value) {
        addComponent(new HeaderItem("set " + key + " " + value + ";", this));
    }

    public void addJar(String jar) {
        addComponent(new HeaderItem("register " + jar + ";", this));
    }

    public void addFunc(String name, String expr) {
        addComponent(new HeaderItem("define " + name + " " + expr + ";", this));
    }

    public Operator newOperator(String opText) {
        Operator op = new Operator();
        op.setText(opText);
        reParseOp(op);
        addOperator(op);
        return op;
    }

    public void addOperator(Operator op) {
        op.setQuery(this);
        components.add(op);
        firePropertyChange(CHILD, null, op);
    }

    public void removeOperator(Operator op) {
        if (parser != null)
            parser.removeOpState(op);
        components.remove(op);

        // make a copy of the output
        List<Link> outputs = new LinkedList<Link>(op.getOutputs());

        op.removeAllInputs();
        op.removeAllOutputs();

        // reparse downstream from deleted operator
        for (Link output : outputs)
            output.getTarget().reParse();

        firePropertyChange(CHILD, op, null);
    }

    public void addLink(Operator source, Operator target) throws IOException {
        if (linkCreatesCycle(source, target))
            throw new IOException(
                    "Attempt to create a cycle in the operator graph");
        Link link = new Link(source, target);
        firePropertyChange(LINK, null, link);
    }

    // would the addition of a link from "source" to "target" create a cycle in
    // the operator graph?
    public boolean linkCreatesCycle(Operator source, Operator target) {
        // see whether "source" is reachable from "target"
        return pathExistsBetween(target, source);
    }

    public boolean pathExistsBetween(Operator a, Operator b) {
        if (a == b)
            return true;

        for (Link link : a.getOutputs()) {
            if (pathExistsBetween(link.getTarget(), b))
                return true;
        }

        return false;
    }

    public void removeLink(Link link) {
        link.getSource().removeOutput(link);
        link.getTarget().removeInput(link);
        firePropertyChange(LINK, link, null);
    }

    // get a list of all operators that are part of this query
    public List<Operator> getOperators() {
        List<Operator> ops = new LinkedList<Operator>();
        for (QueryComponent component : components) {
            if (component instanceof Operator)
                ops.add((Operator) component);
        }
        return ops;
    }

    // get a list of all header items that are part of this query
    public List<HeaderItem> getHeaderItems() {
        List<HeaderItem> items = new LinkedList<HeaderItem>();
        for (QueryComponent component : components) {
            if (component instanceof HeaderItem)
                items.add((HeaderItem) component);
        }
        return items;
    }

    // get a list of operators that have no input links (other than Load
    // operators)
    public List<Operator> getDanglingOperators() {
        List<Operator> danglingOps = new LinkedList<Operator>();
        for (QueryComponent component : components) {
            if (component instanceof Operator) {
                Operator op = (Operator) component;

                String name = op.getName();
                if (name != null && name.equals("Load"))
                    continue;

                if (op.getInputs().isEmpty())
                    danglingOps.add(op);
            }
        }
        return danglingOps;
    }

    public void save(IProgressMonitor monitor) {
        try {
            BufferedWriter writer = new BufferedWriter(new FileWriter(file));
            dumpQuery(writer);
            writer.close();
        } catch (IOException e) {
            LoggerUtil.logError("Error in saving file.", e);
            monitor.setCanceled(true);
        }
        monitor.done();
    }

    private void padHeader() {
        int numHeaderItems = getHeaderItems().size();
        while (numHeaderItems < MAX_HEADER_ITEMS) {
            this.addComponent(new HeaderItem("", this));
            numHeaderItems++;
        }
    }

    private void dumpQuery(BufferedWriter writer) throws IOException {

        // keep a list of operators, which we will write at the end
        Set<Operator> toDump = new HashSet<Operator>();

        // first write header stuff (settings, registered jars, registered
        // funcs)
        for (QueryComponent component : components) {
            if (component instanceof Operator) {
                toDump.add((Operator) component);
            } else if (component instanceof HeaderItem) {
                writeHeaderItem(writer, (HeaderItem) component);
            }
        }
        writer.write("\n"); // extra line to separate header from query
                            // operators

        // the opeartors we need to dump:

        // then write the operators, in topological orders
        while (!toDump.isEmpty()) {
            // find an operator with no non-dumped inputs
            for (Operator dumpMe : toDump) {
                for (Link inputConn : dumpMe.getInputs()) {
                    Operator input = inputConn.getSource();
                    if (toDump.contains(input)) {
                        dumpMe = null;
                        break;
                    }
                }
                if (dumpMe != null) {
                    writer.write(dumpMe.getText());
                    writer.write("\n");
                    toDump.remove(dumpMe);
                    break;
                }
            }
        }

    }

    private void writeHeaderItem(BufferedWriter writer, HeaderItem item)
            throws IOException {
        String str = item.getString();
        if (str != null && str.length() > 0) {
            writer.write(str);
            writer.write("\n");
        }
    }

    public List getComponents() {
        return components;
    }

    public File getFile() {
        return file;
    }

}
